/*
 * 文 件 名:  JavaSever.java
 * 版    权:  Archermind.NANJING 2014-9-3,  All rights reserved
 * 描    述:  <描述>
 * 修 改 人:  Administrator
 * 修改时间:  2014-9-3   上午11:20:27
 * 跟踪单号:  <跟踪单号>
 * 修改单号:  <修改单号>
 * 修改内容:  <修改内容>
 */
package com.demo.netty.mqtt;

import com.demo.netty.util.Config;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * MQTT JAVA 接收和publish 端
 *
 * @author lpf
 * @version [版本号, 2014-9-3]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public final class MqttJavaSeverPublish implements MqttCallback {
    private static final Logger logger = Logger.getLogger(MqttJavaSeverPublish.class);

    // 订阅号 关键字
    //String topic = "lpf";

    private String tmpDir = System.getProperty("java.io.tmpdir");

    private Object waiter = new Object();

    private MqttAsyncClient client = null;

    private MqttConnectOptions conOpt = null;

    private boolean state = false;

    private MqttBean mb;

    private static MqttJavaSeverPublish publish;

    private MqttJavaSeverPublish() {
        mb = new MqttBean();
        //mb.setTopic("07777");
        mb.setClean(Boolean.parseBoolean(Config.getProperty("mqtt_clean")));
        mb.setUserName(Config.getProperty("mqtt_userName"));
        mb.setPassWord(Config.getProperty("mqtt_passWord"));
        mb.setBrokerUrl(Config.getProperty("mqtt_brokerUrl"));
        mb.setClientId(Config.getProperty("mqtt_clientId"));

    }

    public static MqttJavaSeverPublish getPublish() {
        if (null == publish) {
            publish = new MqttJavaSeverPublish();
            publish.initConn();
            publish.connector();
        }
        return publish;
    }

    public static void main(String[] args) {


        MqttBean mb = new MqttBean();
        mb.setTopic("2015/bigdata");
        mb.setClean(Boolean.parseBoolean(Config.getProperty("mqtt_clean")));
        mb.setUserName(Config.getProperty("mqtt_userName"));
        mb.setPassWord(Config.getProperty("mqtt_passWord"));
        mb.setBrokerUrl("tcp://58.213.47.166:8905");
        mb.setClientId("BIGDATA_CLIENT");
        mb.setQos(2);

        MqttJavaSeverPublish publs = MqttJavaSeverPublish.getPublish();

        while (true) {
            String time = new SimpleDateFormat("yyyy-mm-dd hh:mm:ss").format(new Date());
            mb.setBody(( time+ " the time come from server!").getBytes());
            publs.publish(mb);

            try {
                Thread.sleep(3 * 1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                logger.error(e.getMessage());
            }
        }
        //publs.close();
    }

    private void initConn() {
        if (null == mb) {
            return;
        }
        try {

            MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(
                    tmpDir);
            conOpt = new MqttConnectOptions();
            // 心跳时间 30s
            conOpt.setKeepAliveInterval(50);
            conOpt.setCleanSession(mb.isClean());
            if (mb.getPassWord() != null) {
                conOpt.setPassword(mb.getPassWord().toCharArray());
            }
            if (mb.getUserName() != null) {
                conOpt.setUserName(mb.getUserName());
            }
            client = new MqttAsyncClient(mb.getBrokerUrl(), mb.getClientId(),
                    dataStore);
            // Set this wrapper as the callback handler
            client.setCallback(this);

        } catch (MqttException e) {
            // TODO Auto-generated catch block
            logger.error(e.getMessage());
        }

        //Publisher pub = new Publisher();
        // pub.doPublish(topic, qos, msg.getBytes());
    }

    private void connector() {
        MqttConnector con = new MqttConnector();
        con.doConnect();
    }

    public void publish(MqttBean mqtt) {
        while (true) {
            try {
                // 必须等连接成功以后,才能publish 
                if (state) {
                    Publisher pub = new Publisher();
                    pub.doPublish(mqtt.getTopic(),
                            mqtt.getQos(),
                            mqtt.getBody());
                    waitForStateChange(200);
                    break;
                }

                waitForStateChange(200);

            } catch (Exception e) {
                logger.error(e.getMessage());
            }
        }

    }

    //    
    public void close() {
        try {
            Disconnector disc = new Disconnector();
            disc.doDisconnect();
        } catch (Exception e) {
            logger.error(e.getMessage());
            // TODO: handle exception
        }
    }

    /**
     * Wait for a maximum amount of time for a state change event to occur
     *
     * @param maxTTW maximum time to wait in milliseconds
     * @throws MqttException
     */
    private void waitForStateChange(int maxTTW) throws MqttException {
        synchronized (waiter) {

            try {
                waiter.wait(maxTTW);
            } catch (InterruptedException e) {

                logger.error(e.getMessage());
            }

        }
    }

    /**
     * Connect in a non-blocking way and then sit back and wait to be
     * notified that the action has completed.
     */
    public class MqttConnector {

        public MqttConnector() {
        }

        public void doConnect() {
            // Connect to the server
            // Get a token and setup an asynchronous listener on the token which
            // will be notified once the connect completes
            //log("Connecting to " + brokerUrl + " with client ID "
            //        + client.getClientId());

            IMqttActionListener conListener = new IMqttActionListener() {
                public void onSuccess(IMqttToken asyncActionToken) {
                    //log("Connected");
                    state = true;
                    carryOn();
                }

                public void onFailure(IMqttToken asyncActionToken,
                                      Throwable exception) {
                    //ex = exception;
                    state = false;
                    //log("connect failed" + exception);
                    //carryOn();
                }

                public void carryOn() {
                    synchronized (waiter) {
                        //donext = true;
                        waiter.notifyAll();
                    }
                }
            };

            try {
                // Connect using a non-blocking connect
                client.connect(conOpt, "Connect sample context", conListener);
            } catch (MqttException e) {
                logger.error(e.getMessage());
                // If though it is a non-blocking connect an exception can be
                // thrown if validation of parms fails or other checks such
                // as already connected fail.
                //state = ERROR;
                //donext = true;
                // ex = e;
            }
        }
    }

    public class Publisher {
        public void doPublish(String topicName, int qos, byte[] payload) {
            // Send / publish a message to the server
            // Get a token and setup an asynchronous listener on the token which
            // will be notified once the message has been delivered
            MqttMessage message = new MqttMessage(payload);
            message.setQos(qos);

            // String time = new Timestamp(System.currentTimeMillis()).toString();

            // Setup a listener object to be notified when the publish completes.
            //
            IMqttActionListener pubListener = new IMqttActionListener() {
                public void onSuccess(IMqttToken asyncActionToken) {
                    //System.out.println(asyncActionToken.getMessageId());
                    //carryOn();
                }

                public void onFailure(IMqttToken asyncActionToken,
                                      Throwable exception) {
                    System.out.println("onFailure");
                    //carryOn();
                }

                public void carryOn() {
                    synchronized (waiter) {
                        waiter.notifyAll();
                    }
                }
            };

            try {
                // Publish the message
                client.publish(topicName,
                        message,
                        "Pub sample context",
                        pubListener);
            } catch (MqttException e) {
                logger.error(e.getMessage());
            }
        }
    }

    /**
     * Disconnect in a non-blocking way and then sit back and wait to be
     * notified that the action has completed.
     */
    public class Disconnector {
        public void doDisconnect() {
            // Disconnect the client
            //log("Disconnecting");

            IMqttActionListener discListener = new IMqttActionListener() {
                public void onSuccess(IMqttToken asyncActionToken) {
                    //log("Disconnect Completed");
                    //state = DISCONNECTED;
                    //carryOn();
                }

                public void onFailure(IMqttToken asyncActionToken,
                                      Throwable exception) {
                    System.out.println("disconn onfailure");
                    //ex = exception;
                    //state = ERROR;
                    //log("Disconnect failed" + exception);
                    //carryOn();
                }

                public void carryOn() {
                    synchronized (waiter) {
                        //donext = true;
                        waiter.notifyAll();
                    }
                }
            };

            try {
                client.disconnect("Disconnect sample context", discListener);
            } catch (MqttException e) {
                logger.error(e.getMessage());
                //state = ERROR;
                //donext = true;
                // ex = e;
            }
        }
    }

    // 断网重连
    @Override
    public void connectionLost(Throwable arg0) {
        try {
            state = false;
            while (true) {
                System.out.println("reconn start...");
                MqttConnector con = new MqttConnector();
                con.doConnect();

                if (state) {
                    //    Publisher pub = new Publisher();
                    //    pub.doPublish(mb.getTopic(), mb.getQos(), mb.getBody());
                    break;
                }

                waitForStateChange(5000);
            }

        } catch (Exception e) {
            logger.error(e.getMessage());
            // TODO: handle exception
        }
        //        System.out.println("dsfadfdas");
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken arg0) {
        try {
            //System.out.println(arg0.getClient().getClientId());
            logger.info("send client id " + arg0.getClient().getClientId()
                    + " ok.");
            //            for(String tp : arg0.getTopics())
            //            {
            //                System.out.println(tp);
            //            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            logger.error(e.getMessage());
        }
    }

    @Override
    public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
        String time = new Timestamp(System.currentTimeMillis()).toString();
        logger.info("Time:\t" + time + "  Topic:\t" + arg0
                + "  Message:\t" + new String(arg1.getPayload()) + "  QoS:\t"
                + arg1.getQos());
        //PMBean pmb = new PMBean();
        // if (pmb.format(arg0))
        //{
        //   ProductBean.getProduct().addQueue(pmb);
        //}
        //else
        //{
        //     Servers.logger.error("msg is paramt error");
        // /}
    }
}
